/*
 * Copyright 2006-2023 ahtelit.com, All Rights Reserved.
 */

package com.telit.firecommand.config.mqtt;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * MQTT客户端管理类,如果客户端非常多后续可入redis缓存
 *
 * @author yuezhuang
 * @since 2023-04-28
 */

@Slf4j
@Component
public class MqttClientManager {

    @Autowired
    private MqttCallBackContext mqttCallBackContext;

    /**
     * 存储MQTT客户端
     */
    public static Map<String, MqttClient> MQTT_CLIENT_MAP = new ConcurrentHashMap<>();

    /**
     * 获取MQTT客户端
     */
    public static MqttClient getMqttClientById(String clientId) {
        return MQTT_CLIENT_MAP.get(clientId);
    }

    /**
     * 删除MQTT客户端
     */
    public static void removeMqttClientById(String clientId) {
        MQTT_CLIENT_MAP.remove(clientId);
    }

    /**
     * 创建mqtt客户端
     *
     * @param mqttBroker     服务端地址
     * @param clientId       客户端ID
     * @param subscribeTopic 订阅主题，可为空
     * @param userName       用户名，可为空
     * @param password       密码，可为空
     */
    public void createMqttClient(String mqttBroker, String clientId, List<String> subscribeTopic, String userName,
                                 String password) {
        MqttClient client = getMqttClientById(clientId);
        if (client != null) {
            return;
        }

        MemoryPersistence persistence = new MemoryPersistence();
        try {
            client = new MqttClient(mqttBroker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            if (StrUtil.isNotEmpty(userName)) {
                connOpts.setUserName(userName);
            }
            if (StrUtil.isNotEmpty(password)) {
                connOpts.setPassword(password.toCharArray());
            }
            connOpts.setCleanSession(true);
            if (CollectionUtil.isNotEmpty(subscribeTopic)) {
                AbsMqttCallBack callBack = mqttCallBackContext.getCallBack(clientId);
                if (null == callBack) {
                    callBack = mqttCallBackContext.getCallBack("defaultMqttCallBack");
                }
                callBack.setClientId(clientId);
                callBack.setConnectOptions(connOpts);
                client.setCallback(callBack);
            }

            //连接mqtt服务端broker
            client.connect(connOpts);

            //订阅主题
            if (CollectionUtil.isNotEmpty(subscribeTopic)) {
                client.subscribe(subscribeTopic.toArray(new String[0]));
            }

            MQTT_CLIENT_MAP.putIfAbsent(clientId, client);

            log.info("mqtt connect success：{}, {}, {}, {}", mqttBroker, clientId, subscribeTopic, userName);
        } catch (MqttException e) {
            log.error("Create mqttClient failed!", e);
        }
    }

    /**
     * 断开连接
     */
    public static void disConnect(String clientId) {
        try {
            MqttClient mqttClient = getMqttClientById(clientId);
            if (mqttClient != null) {
                mqttClient.disconnect();
            }

            log.info("mqtt disConnect success：{}", clientId);
        } catch (MqttException e) {
            log.error("mqtt disConnect error：", e);
        }
    }


    /**
     * 订阅主题
     */
    public static void subscribe(String clientId, List<String> subscribeTopic) {
        try {
            MqttClient mqttClient = getMqttClientById(clientId);
            if (mqttClient != null) {
                mqttClient.subscribe(subscribeTopic.toArray(new String[0]));
            }
            
            log.info("mqtt subscribe success：{}, {}", clientId, subscribeTopic);
        } catch (MqttException e) {
            log.error("mqtt subscribe error：", e);
        }
    }

}
